package smart.rpc.client;

import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import smart.rpc.common.SmartRPCRequest;
import smart.rpc.common.SmartRPCResponse;
import smart.rpc.exception.SmartRPCException;

import java.util.*;
import java.util.concurrent.*;

/**
 * Created By gao_e on 2020/3/10 15:06
 * RPC 调用的 客户端
 */
public class StartNettyClientConnectTask implements Runnable {

    // 当前服务作为客户端，那么针对每一个不同的服务端都会用一个独立的线程来维护一个netty NIO 长连接，
    // 用于当前服务请求远程服务做数据交互
    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    // 主机地址
    private String host;
    // 服务端口
    private int port;

    // 判断是否存在请求任务的信号量
    private Semaphore requestSemaphore = new Semaphore(0);

//    private Semaphore requestSemaphoreNew = new Semaphore(0);

    // 用于线程间交互，存储请求对象的临时队列
    private ConcurrentLinkedDeque<SmartRPCRequest> smartRPCRequests = new ConcurrentLinkedDeque();
    // 需要保证线程安全
//    private List<SmartRPCRequest> smartRPCRequests = Collections.synchronizedList(requestList);

    // TODO 待优化,避免一次取一个
    public SmartRPCRequest handedARequest() {
        try {
            // 消耗一个信号量
            requestSemaphore.acquire(1);
        } catch (InterruptedException e) {
            throw new SmartRPCException("WRITE REQUEST 2 SERVER ERROR , e = " + e);
        }
        // TODO 线程安全问题？
        // 释放一个请求任务
        SmartRPCRequest firstRequest = smartRPCRequests.pollFirst();
        return firstRequest;
    }

    // TODO 待优化方案, 需要保证线程安全
    public List<SmartRPCRequest> handedARequestList() {
//        smartRPCRequests.
        // 释放一个请求任务
        return null;
    }

    /**
     * 有一个新的请求需要RPC远程处理
     */
    public void addedRequest(SmartRPCRequest smartRPCRequest) {
        // 添加到请求队列池
        smartRPCRequests.addLast(smartRPCRequest);
        // 添加一个响应信号，默认是灭灯(信号量为0)状态
        responseSemphoreMap.put(smartRPCRequest.getRequestId(), new Semaphore(0));
        // 释放一个信号，告知IO线程执行远程请求写操作
        requestSemaphore.release(1);
    }

    // 请求id,是否已响应信号
    private Map<String, Semaphore> responseSemphoreMap = new ConcurrentHashMap<>(512);

    // 请求id,响应结果
    private Map<String, SmartRPCResponse> smartRPCResponseMap = new ConcurrentHashMap<>(512);

    public void receiveAResponse(SmartRPCResponse response) {
        // 响应添加到响应队列
        smartRPCResponseMap.put(response.getRequestId(), response);
        // 释放已响应信号
        responseSemphoreMap.get(response.getRequestId()).release(1);
    }

    public Object synchGetResult(String requestId) {
        Semaphore semaphore = responseSemphoreMap.get(requestId);
        boolean getedData = false;
        try {
            getedData = semaphore.tryAcquire(1, 10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new SmartRPCException("SERVER UNKNOWN ERROR");
        }
        if (!getedData)
            throw new SmartRPCException("REQUEST WAIT TIME OUT 3 SECONDS ERROR");
        SmartRPCResponse smartRPCResponse = smartRPCResponseMap.get(requestId);
        // 移除引用，避免内存泄露
        smartRPCResponseMap.remove(requestId);
        responseSemphoreMap.remove(requestId);
        if(smartRPCResponse == null || !smartRPCResponse.isSuccess())
            throw new SmartRPCException("Request unknown error , msg = " + smartRPCResponse.getMsg());
        // 对于结果,需要进一步转换,转换为方法的返回结果类型
        Object result = smartRPCResponse.getData();
        String dataJson = JSONObject.toJSONString(result);
        Object data = JSONObject.parseObject(dataJson, smartRPCResponse.getResultCls());
        return data;
    }

    public StartNettyClientConnectTask(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void addClientConnectRunnableTask2InnerThreadPoolAndExecuteSelf() {
        executorService.execute(this);
    }

    @Override
    public void run() {
        try {
            // 连接服务端
            connect();
        } catch (Exception e) {
            System.out.println(e);
            throw new SmartRPCException("CONNECTINT SERVER ERROR , server host = " + host + " , port = " + port);
        }
    }

    private void connect() throws Exception {
        // 开启 netty NIO , 连接远程server
        EventLoopGroup group = new NioEventLoopGroup(4);
        try {
            Bootstrap bootstrap = new Bootstrap();
            ChannelInitializer channelInitializer = new NettyClientChannelInitializer(this);
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.AUTO_READ, true)
                    .handler(channelInitializer)
                    .option(ChannelOption.SO_KEEPALIVE, true);;
            // 发起同步连接操作
            ChannelFuture future = bootstrap.connect(host, port).sync();
            // 直接写出
            // new WriteRequestTask(this).execute(future);
            // 同步等待客户端链路关闭
            future.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放NIO线程组
            group.shutdownGracefully();
        }
    }

    @Deprecated
    private static class WriteRequestTask implements Runnable {
        public void execute(ChannelFuture future){
            this.future = future;
            executorService.execute(this);
        }
        private ChannelFuture future;
        private StartNettyClientConnectTask startNettyClientConnectTask;
        public  WriteRequestTask(StartNettyClientConnectTask startNettyClientConnectTask){
            this.startNettyClientConnectTask = startNettyClientConnectTask;
        }
        private ExecutorService executorService = Executors.newFixedThreadPool(1);
        @Override
        public void run() {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        System.out.println("...............开始执行请求写出操作...............");
                        SmartRPCRequest request = startNettyClientConnectTask.handedARequest();
                        String requestJson = JSONObject.toJSONString(request);
                        future.channel().writeAndFlush("requestJson");
                    }
                }
            });
        }
    }

}
